package cn.pengpeng.day05.firstMR;

import java.io.IOException;
import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 提交咱们的任务
 * @author root
 */
public class Driver {
	public static void main(String[] args) throws Exception {
		//什么都没有配置，怎么知道yarn集群在哪里      怎么知道hdfs集群在哪里   
		//加载集群上面的配置文件
		Configuration conf = new Configuration(); //加载集群上的配置文件
		conf.set("fs.defaultFS", "hdfs://hadoop01:9000");
		Job job = Job.getInstance(conf);
		
		//设置job的map和reduce是哪一个，并且设置是哪一做任务提交
		job.setMapperClass(MapTask.class);
		job.setReducerClass(ReduceTask.class);
		//job.setCombinerClass(MyCombiner.class);
		job.setCombinerClass(ReduceTask.class);
		job.setJarByClass(Driver.class);
		
		//设置输出类型
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		job.setNumReduceTasks(2);
		
		//设置输入和输出目录
		FileInputFormat.addInputPath(job, new Path("/wc.txt"));
		FileOutputFormat.setOutputPath(job, new Path("/wordcount/wc-output"));
		
		//判断文件夹是否存在（输出目录是否存在）
		FileSystem fs = FileSystem.get(conf);
		if(fs.exists(new Path("/wordcount/wc-output"))){
			fs.delete(new Path("/wordcount/wc-output"), true);
		}
		
		// 提交之后会监控运行状态
		boolean completion = job.waitForCompletion(true);
		System.out.println(completion?"程序执行完毕，没毛病！！！":"程序有问题，程序出bug了，赶紧加班调试！！！");
		
	}

}
